"""
Copyright (c) 2022 Huawei Technologies Co.,Ltd.

openGauss is licensed under Mulan PSL v2.
You can use this software according to the terms and conditions of the Mulan PSL v2.
You may obtain a copy of Mulan PSL v2 at:

          http://license.coscl.org.cn/MulanPSL2

THIS SOFTWARE IS PROVIDED ON AN "AS IS" BASIS, WITHOUT WARRANTIES OF ANY KIND,
EITHER EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO NON-INFRINGEMENT,
MERCHANTABILITY OR FIT FOR A PARTICULAR PURPOSE.
See the Mulan PSL v2 for more details.
"""
"""
Case Type   : Logical_Replication
Case Name   : 有主键且replica identity属性为default 的分区表逻辑解码
Create At   : 2023/07/15
Owner       : @xiaqi14
Description :
    1.修改参数
    2.配置初始用户白名单
    3.创建分区表
    4.创建逻辑复制槽
    5.主机和备机分别开启逻辑复制槽
    6.执行DML操作
    7.停止逻辑复制槽
    8.删除逻辑复制槽
    9.资源清理
    10.恢复参数，恢复pg_hba.conf文件
Expect      :
    1.修改参数成功
    2.配置成功
    3.创建成功
    4.创建成功
    5.开启成功
    6.执行DML操作成功
    7.停止成功
    8.删除成功
    9.清理成功
    10.恢复成功
History     :
"""
import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread

primary_sh = CommonSH('PrimaryDbUser')


@unittest.skipIf(1 == primary_sh.get_node_num(), '单机环境不执行')
class LogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.standby_node = Node('Standby1DbUser')
        self.sh_standby = CommonSH('Standby1DbUser')
        self.constant = Constant()
        self.common = Common()
        self.table_name = 't_logical_replication_0012'
        self.slot_name_1 = 'slot_logical_replication_0012_1'
        self.slot_name_2 = 'slot_logical_replication_0012_2'
        self.logfile_1 = os.path.join(macro.DB_BACKUP_PATH,
                                      'logfile_logical_replication_0012_1.log')
        self.logfile_2 = os.path.join(macro.DB_BACKUP_PATH,
                                      'logfile_logical_replication_0012_2.log')
        self.copy_path = os.path.join(macro.DB_BACKUP_PATH,
                                      'copy_table_0006.csv')
        self.config_slot_log = 'enable_slot_log'
        self.slot_log_default = self.common.show_param(self.config_slot_log)
        self.config_wal_level = 'wal_level'
        self.wal_level_default = self.common.show_param(self.config_wal_level)

    def test_logical_replication(self):
        text = "-----step1:修改参数;expect:修改参数成功-----"
        self.logger.info(text)
        if 'on' != self.slot_log_default:
            guc_result = primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}=on')
            self.assertTrue(guc_result, '执行失败' + text)
        show_msg = self.common.show_param(self.config_slot_log)
        self.assertEquals(show_msg, 'on', '执行失败' + text)

        if 'logical' != self.wal_level_default:
            guc_result = primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = primary_sh.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_msg, 'logical', '执行失败' + text)

        text = "-----step2:配置初始用户白名单;expect:配置成功-----"
        self.logger.info(text)
        param_1 = f'local replication {self.primary_node.ssh_user}  trust'
        param_2 = f'host  replication {self.primary_node.ssh_user} ' \
            f'127.0.0.1/32 trust'
        param_3 = f'host  replication {self.primary_node.ssh_user} ' \
            f'::1/128 trust'
        result_1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step3:创建分区表;expect:创建成功-----"
        self.logger.info(text)
        create_cmd = f'''drop table if exists {self.table_name};
            create table {self.table_name}
            (
                col_integer integer ,
                col_bigint bigint, 
                col_smallint smallint, 
                col_tinyint tinyint,
                col_serial serial,
                col_smallserial smallserial,
                col_bigserial bigserial,
                col_float float,
                col_double double precision,
                col_date date,
                col_time time without time zone,
                col_timestamp timestamp without time zone,
                col_char char(10),
                col_varchar varchar(20),
                col_text text,
                col_blob blob,
                col_bytea bytea
            )partition by range(col_integer)(
                partition p1 values less than(100),
                partition p2 values less than(200),
                partition p3 values less than(300),
                partition p6 values less than(maxvalue));
            alter table {self.table_name} replica identity default;
            alter table t_logical_replication_0012 
            add constraint par_p_k_0012 primary key(col_integer);'''
        create_msg = primary_sh.execut_db_sql(create_cmd, dbname='postgres')
        self.logger.info(create_msg)
        self.assertEquals(4, create_msg.count(
            self.constant.CREATE_TABLE_SUCCESS), '执行失败:' + text)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS,
                      create_msg, '执行失败:' + text)
        self.assertEquals(3, create_msg.count(self.constant.ALTER_TABLE_MSG),
                          '执行失败:' + text)

        text = "-----step4:创建逻辑复制槽;expect:创建成功-----"
        self.logger.info(text)
        slot_names = [self.slot_name_1, self.slot_name_2]
        for slot_name in slot_names:
            create_slot_msg = primary_sh.exec_pg_recvlogical(
                '--create', 'postgres', slot_name,
                self.primary_node.ssh_user, '-s 2')
            self.logger.info(create_slot_msg)
        text = "-----验证创建逻辑复制槽成功;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%slot_logical_replication_0012%';"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1 and self.slot_name_2,
                      sql_msg, '执行失败:' + text)

        text = "-----step5:主机和备机分别开启逻辑复制槽;expect:开启成功-----"
        self.logger.info(text)
        thread_1 = ComThread(primary_sh.exec_pg_recvlogical,
                             args=('--start', 'postgres', self.slot_name_1,
                                   self.primary_node.ssh_user,
                                   f'-f {self.logfile_1} '
                                   f'-o skip-empty-xacts=true -F 1 -s 2',))
        thread_1.setDaemon(True)
        thread_1.start()
        thread_1.join(10)
        self.logger.info(thread_1.get_result())
        text = "-----验证主机开启逻辑复制槽成功;expect:开启成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_1}%';;"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertEquals('t', sql_msg.splitlines()[-2].strip(),
                          '执行失败:' + text)
        thread_2 = ComThread(self.sh_standby.exec_pg_recvlogical,
                             args=('--start', 'postgres', self.slot_name_2,
                                   self.primary_node.ssh_user,
                                   f'-f {self.logfile_2} '
                                   f'-o skip-empty-xacts=true -F 1 -s 2',))
        thread_2.setDaemon(True)
        thread_2.start()
        thread_2.join(10)
        self.logger.info(thread_2.get_result())
        text = "-----验证备机开启逻辑复制槽成功;expect:开启成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_2}%';;"
        sql_msg = self.sh_standby.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertEquals('t', sql_msg.splitlines()[-2].strip(),
                          '执行失败:' + text)

        text = "-----step6:执行DML操作;expect:执行DML操作成功，" \
               "解码成功，update解码为delete和insert-----"
        self.logger.info(text)
        sql_cmd = f'''insert into {self.table_name} values(10,10,5,25,
            default,default,default,1237.127,123456.1234,
            date '2022-03-22','14:21:21','2022-04-25','测试',
            '测试工程师','西安',empty_blob(),E'\\xDEADBEEF');
            update {self.table_name} set col_integer=180,col_bigint=130,
            col_smallint=124,col_tinyint=125,col_float=113.85,
            col_date='2022-04-01',col_time='18:30:00',
            col_timestamp='2022-03-27',col_char='开发',
            col_varchar='开发工程师',col_text='陕西';
            copy {self.table_name} to '{self.copy_path}';
            delete from {self.table_name};
            copy {self.table_name} from '{self.copy_path}';'''
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.INSERT_SUCCESS_MSG and
                      self.constant.UPDATE_SUCCESS_MSG and
                      self.constant.DELETE_SUCCESS_MSG,
                      sql_msg, "执行失败" + text)
        self.assertEquals(2, sql_msg.count("COPY"), '执行失败:' + text)
        cat_cmd_1 = f"cat {self.logfile_1} "
        self.logger.info(cat_cmd_1)
        cat_msg_1 = self.primary_node.sh(cat_cmd_1).result()
        self.logger.info(cat_msg_1)
        self.assertIn('"op_type":"INSERT"', cat_msg_1.splitlines()[1],
                      "执行失败" + text)
        self.assertIn('"op_type":"DELETE"' and
                      '"old_keys_name":["col_integer"],'
                      '"old_keys_type":["integer"],"old_keys_val":["10"]',
                      cat_msg_1.splitlines()[4],
                      "执行失败" + text)
        self.assertIn('"op_type":"INSERT"',
                      cat_msg_1.splitlines()[5], "执行失败" + text)
        self.assertIn('"op_type":"DELETE"' and
                      '"old_keys_name":["col_integer"],'
                      '"old_keys_type":["integer"],"old_keys_val":["180"]',
                      cat_msg_1.splitlines()[8],
                      "执行失败" + text)
        self.assertIn('"op_type":"INSERT"', cat_msg_1.splitlines()[11],
                      "执行失败" + text)

    def tearDown(self):
        text = "-----step7:停止逻辑复制槽;expect:停止成功-----"
        self.logger.info(text)
        stop_cmd = "ps -ef | grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.logger.info(stop_cmd)
        primary_result = self.primary_node.sh(stop_cmd).result()
        self.logger.info(primary_result)
        standby_result = self.standby_node.sh(stop_cmd).result()
        self.logger.info(standby_result)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name_1}%';"
        primary_active_msg = primary_sh.execut_db_sql(sql_cmd,
                                                      dbname='postgres')
        self.logger.info(primary_active_msg)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name_2}%';"
        standby_active_msg = self.sh_standby.execut_db_sql(sql_cmd,
                                                           dbname='postgres')
        self.logger.info(standby_active_msg)

        text_1 = "-----step8:删除逻辑复制槽;expect:删除成功-----"
        self.logger.info(text_1)
        slot_names = [self.slot_name_1, self.slot_name_2]
        for slot_name in slot_names:
            drop_slot_msg = primary_sh.exec_pg_recvlogical(
                '--drop', 'postgres', slot_name,
                self.primary_node.ssh_user, '-s 2')
            self.logger.info(drop_slot_msg)
        sql_cmd = f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%slot_logical_replication_0012%';"
        sql_msg = primary_sh.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)

        text_2 = "-----step9:资源清理;expect:清理成功-----"
        self.logger.info(text_2)
        drop_cmd = f'''drop table {self.table_name} cascade;'''
        drop_msg = primary_sh.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)
        del_cmd = f'rm -rf {self.logfile_1} {self.copy_path};'
        self.logger.info(del_cmd)
        primary_del_res = self.primary_node.sh(del_cmd).result()
        self.logger.info(primary_del_res)
        del_cmd = f'rm -rf {self.logfile_2};'
        self.logger.info(del_cmd)
        standby_del_res = self.standby_node.sh(del_cmd).result()
        self.logger.info(standby_del_res)
        file_cmd = f'''if [[ -e {self.logfile_1} && -e {self.copy_path} ]];
            then echo "exists"; else echo "not exists"; fi'''
        self.logger.info(file_cmd)
        primary_file_msg = self.primary_node.sh(file_cmd).result()
        self.logger.info(primary_file_msg)
        file_cmd = f'''if [ -f {self.logfile_2} ];
                    then echo "exists"; else echo "not exists"; fi'''
        self.logger.info(file_cmd)
        standby_file_msg = self.standby_node.sh(file_cmd).result()
        self.logger.info(standby_file_msg)

        text_3 = "-----step10:恢复参数，恢复pg_hba.conf文件，expect:恢复成功-----"
        self.logger.info(text_3)
        slot_log_current = self.common.show_param(self.config_slot_log)
        if slot_log_current != self.slot_log_default:
            primary_sh.execute_gsguc(
                'reload', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_slot_log}={self.slot_log_default}')
        show_slot_log = self.common.show_param(self.config_slot_log)

        wal_level_current = self.common.show_param(self.config_wal_level)
        if wal_level_current != self.wal_level_default:
            primary_sh.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}={self.wal_level_default}')
            restart_msg = primary_sh.restart_db_cluster()
            self.logger.info(restart_msg)
        show_wal_level = self.common.show_param(self.config_wal_level)

        param_1 = f'local replication {self.primary_node.ssh_user} '
        param_2 = f'host replication {self.primary_node.ssh_user} 127.0.0.1/32 '
        param_3 = f'host  replication {self.primary_node.ssh_user} ::1/128 '
        result_1 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = primary_sh.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)

        self.assertEquals('f', primary_active_msg.splitlines()[-2].strip(),
                          '执行失败:' + text)
        self.assertEquals('f', standby_active_msg.splitlines()[-2].strip(),
                          '执行失败:' + text)
        self.assertIn("(0 rows)", sql_msg, '执行失败:' + text_1)
        self.assertIn(self.constant.TABLE_DROP_SUCCESS,
                      drop_msg, '执行失败' + text_2)
        self.assertEqual("not exists", primary_file_msg, '执行失败' + text_2)
        self.assertEqual("not exists", standby_file_msg, '执行失败' + text_2)
        self.assertEquals(show_slot_log, self.slot_log_default,
                          '执行失败' + text_3)
        self.assertEquals(show_wal_level, self.wal_level_default,
                          '执行失败' + text_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text_3)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
